package com.hsc.www.rawTest.事务;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ConfirmSend3 {
    /**
     * 队列名
     */
    private static final String QUEUE_NAME = "test_queue_confirm1";
    /**
     * exchange
     */
    private static final String EXCHANGE_NAME = "test_queue_exchange";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 创建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setHost("172.172.172.11");
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPort(5672);
        // 获取 connection
        Connection connection = connectionFactory.newConnection();
        // 从连接开一个通道
        Channel channel = connection.createChannel();
        // 声明一个队列
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, null);

        Map queueArgs = new HashMap();
        queueArgs.put("x-overflow", "drop-head");
        queueArgs.put("x-max-length", 10000);
        queueArgs.put("x-dead-letter-exchange", "deadLetterExchange");
        queueArgs.put("x-dead-letter-routing-key", "deadLetterRoutingKey");
        channel.queueDeclare(QUEUE_NAME, false, false, false, queueArgs);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, QUEUE_NAME);

        // 开始confirm模式
        channel.confirmSelect();
        // 未确认的消息列表
        SortedSet<Long> waitAckMessageDeliveryTags = Collections.synchronizedSortedSet(new TreeSet<>());
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------ack!-----------deliveryTag=" + deliveryTag + "  multiple=" + multiple);
                if (multiple) {
                    waitAckMessageDeliveryTags.headSet(deliveryTag + 1).clear();
                } else {
                    waitAckMessageDeliveryTags.remove(deliveryTag);
                }
            }

            //
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------no ack!-----------deliveryTag=" + deliveryTag + "  multiple=" + multiple);
                if (multiple) {
                    waitAckMessageDeliveryTags.headSet(deliveryTag + 1).clear();
                } else {
                    waitAckMessageDeliveryTags.remove(deliveryTag);
                }
            }
        });
        //如果消息从交换器发送到对应队列失败时触（比如根据发送消息时指定的routingKey找不到队列时会触发）
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.err.println("-------handleReturn-----------" + new String(body, StandardCharsets.UTF_8));
            }
        });

        new Thread(() -> {
            //delivery-mode有两个值：1表示非持久化，2表示持久化消息.非持久化的消息在服务宕机的时候会丢失数据，但是由于不需要磁盘io，尽可能地降低消息投递的延迟性，性能较高。
            //持久化的消息安全性较高，尽管服务宕机，数据也不会丢失，但是在投递消息的过程中需要发生磁盘io，性能相对纯内存投递的方式低.
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().deliveryMode(2).build();
            //当mandatory为true 时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者;
            //当mandatory为false时，出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者;
            boolean mandatory = true;
            try {
                for (int i = 0; i < 10; i++) {
                    long deliveryTag = channel.getNextPublishSeqNo();
                    String message = "消息" + deliveryTag;
                    channel.basicPublish(EXCHANGE_NAME, QUEUE_NAME, true, basicProperties, message.getBytes());
                    waitAckMessageDeliveryTags.add(deliveryTag);
                    System.out.println("发送：" + message);
                    TimeUnit.MILLISECONDS.sleep(Math.round(Math.random() * 2));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

    }
}
